package com.ccb.channel.zj;

import org.apache.log4j.Logger;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;

import com.alibaba.fastjson.JSON;
import com.ccb.filter.Filters;
import com.ccb.filter.RecordFilter;

public class P6Mapper  extends Mapper<Object, Text, Text, Text>{
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(P6Mapper.class);

	private Text group = new Text();
	private Text result = new Text();
	private RecordFilter recordFilter ; 
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		recordFilter  =  Filters.getFilter("p6");
		if(recordFilter == null){
			logger.info("fiter is null");
		}
	}
	private RecordGenerator recordGenerator = new ZJRecordGenerator();

	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("map(Object, Text, Context) - start");
		}

		String content = value.toString().trim();
		if(content.isEmpty())
			return;
		MapperResult mapperResult = recordGenerator.parseData(content);
		
		if(recordFilter != null && !recordFilter.accept(mapperResult.getRecord().getServiceId())){
			if (logger.isDebugEnabled()) {
				logger.debug("map(Object, Text, Context) - end");
			}
			return;
		}
		group.set(mapperResult.getGroup());
		result.set(JSON.toJSONString(mapperResult.getRecord()));
		context.write(group, result);

		if (logger.isDebugEnabled()) {
			logger.debug("map(Object, Text, Context) - end");
		}
	}
}
